package com.smile.realtime.utils

import com.smile.realtime.bean.DauInfo
import io.searchbox.client.config.HttpClientConfig
import io.searchbox.client.{JestClient, JestClientFactory}
import io.searchbox.core.{Bulk, BulkResult, Index, Search, SearchResult}
import org.apache.avro.ipc.reflect.ReflectRequestor.getClient

import java.util


  object MyESUtil {
    private var factory: JestClientFactory = null;

    def getClient: JestClient = {
      if (factory == null) build();
      factory.getObject
    }

    def build(): Unit = {
      factory = new JestClientFactory
      factory.setHttpClientConfig(new
          HttpClientConfig.Builder("http://spider01:9200")
        .multiThreaded(true)
        .maxTotalConnection(20)
        .connTimeout(10000).readTimeout(1000).build())
    }


    //向 ES 中插入数据
    def putIndex(): Unit = {
      //建立连接
      val jest: JestClient = getClient
      //Builder 中的参数，底层会转换为 Json 格式字符串，所以我们这里封装 Document 为样例类
      //当然也可以直接传递 json
      val actorNameList = new util.ArrayList[String]()
      actorNameList.add("zhangsan")
      val index: Index = new Index.Builder(Movie("100", "天龙八部", actorNameList))
        .index("movie_index_5")
        .`type`("movie")
        .id("1")
        .build()
      //execute 的参数类型为 Action，Action 是接口类型，不同的操作有不同的实现类，添加的实现类为 Index
      jest.execute(index)
      //关闭连接
      jest.close()
    }

    //向 ES 中批量插入数据
    //参数 1：批量操作的数据 参数 2：索引名称
    def bulkInsert(sourceList: List[(String,Any)], indexName: String): Unit = {
      if (sourceList != null && sourceList.size > 0) {
        //获取操作对象
        val jest: JestClient = getClient
        //构造批次操作
        val bulkBuild: Bulk.Builder = new Bulk.Builder
        //对批量操作的数据进行遍历
        for ((id,dauInfo) <- sourceList) {
          val index: Index = new Index.Builder(dauInfo)
            .index(indexName)
            .id(id)
            .`type`("_doc")
            .build()
          //将每条数据添加到批量操作中
          bulkBuild.addAction(index)
        }
        //Bulk 是 Action 的实现类，主要实现批量操作
        val bulk: Bulk = bulkBuild.build()
        //执行批量操作 获取执行结果
        val result: BulkResult = jest.execute(bulk)
        //通过执行结果 获取批量插入的数据
        val items: util.List[BulkResult#BulkResultItem] = result.getItems
        println("保存到 ES" + items.size() + "条数")
        //关闭连接
        jest.close()
      }
    }


    //从 ES 中查询数据
    def queryIndex(): Unit = {
      //获取操作对象
      val jest: JestClient = getClient
      //查询常用有两个实现类 Get 通过 id 获取单个 Document，以及 Search 处理复杂查询
      val query =
        """
          |{
          | "query": {
          | "bool": {
          | "must": [
          | {"match": {
          | "name": "red"
          | }}
          | ],
          | "filter": [
          | {"term": { "actorList.name.keyword": "zhang cuishan"}}
          | ]
          | }
          | },
          | "from": 0,
          | "size": 20,
          | "sort": [
          | {
          | "doubanScore": {
          | "order": "desc"
          | }
          | }
          | ],
          | "highlight": {
          | "fields": {
          | "name": {}
          | }
          | }
          |}
 """.stripMargin
      val search: Search = new Search.Builder(query)
        .addIndex("movie_index")
        .build()
      //执行操作
      val result: SearchResult = jest.execute(search)
      //获取命中的结果 sourceType:对命中的数据进行封装，因为是 Json，所以我们用 map 封装
      //注意：一定得是 Java 的 Map 类型
      val rsList: util.List[SearchResult#Hit[util.Map[String, Any], Void]] =
      result.getHits(classOf[util.Map[String, Any]])
      //将 Java 转换为 Scala 集合，方便操作
      import scala.collection.JavaConverters._
      //获取 Hit 中的 source 部分
      val list: List[util.Map[String, Any]] = rsList.asScala.map(_.source).toList
      println(list.mkString("\n"))
      //关闭连接
      jest.close()
    }


    case class Movie(id: String, movie_name: String, actorNameList: java.util.List[String]) {}

    //测试查看效果
    def main(args: Array[String]): Unit = {
      putIndex()
    }
  }



